package com.ts.api.task.org;

import com.gitee.apanlh.util.base.MapUtils;
import com.gitee.apanlh.util.cache.local.CacheFifo;
import com.gitee.apanlh.util.sys.SystemInfo;
import com.gitee.apanlh.util.thread.Sleep;
import com.gitee.apanlh.util.thread.pool.ThreadPoolExecutorUtils;
import net.openhft.affinity.Affinity;

import java.math.BigInteger;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

/**
 *  机构唤醒
 *
 *  @author Pan
 */
//@Component
public class OrgWakeTask implements Runnable {

    /** 自定义渠道唤醒线程池 */
    public static final ThreadPoolExecutor POOL = ThreadPoolExecutorUtils.instanceBlocking(16, 16);
    /** 普通平台唤醒及机构唤醒线程池 */
//    public static final ThreadPoolExecutor NORMAL_POOL = ThreadPoolExecutorUtils.instanceBlocking(16, 16);

    /** 最大cpu核心数 */
    private static final int MAX_AVAILABLE_INT = (Integer.valueOf(SystemInfo.create().getCpuAvailable()) - 1);
    /** 实际分配数-整型 */
    private static final int AVAILABLE_INT = (MAX_AVAILABLE_INT / 2) + 4;
    /** 实际分配数-字符串 */
    private static final String AVAILABLE_STR = String.valueOf(AVAILABLE_INT);
    /** 唤醒任务池 */
    private static final Map<Integer, OrgWakeTask> BIND_MAP = MapUtils.newHashMap(AVAILABLE_INT);

    /** 绑定值 */
    private int bind = 0;
    /** 最大队列10000 */
    private final CacheFifo<String, Runnable> wakeQueue = new CacheFifo(10000);

    /**
     *  默认构造函数
     *  @author Pan
     */
    public OrgWakeTask() {
        super();
        initBindStrategy();
    }

    /**
     *  构造函数
     *  自定义绑定
     *
     *  @author Pan
     */
    public OrgWakeTask(int bindCpu) {
        this.bind = bindCpu;
    }

    /**
     *  初始化绑定策略
     *
     *  @author Pan
     */
    void initBindStrategy() {
        for (int bind = 0; bind <= AVAILABLE_INT; bind++) {
            OrgWakeTask orgWakeTask = new OrgWakeTask(bind);
            orgWakeTask.setBind(bind);
            BIND_MAP.put(bind, orgWakeTask);
            new Thread(() -> orgWakeTask.run()).start();
        }
    }

    public int getBind() {
        return bind;
    }

    public void setBind(int bind) {
        this.bind = bind;
    }

    /**
     *  获取唤醒队列
     *
     *  @author Pan
     *  @return CacheFifo
     */
    public CacheFifo<String, Runnable> getWakeQueue() {
        return this.wakeQueue;
    }

    @Override
    public void run() {
        Affinity.setAffinity(this.getBind());
        Affinity.acquireCore(false);

        //  执行任务
        CacheFifo<String, Runnable> wakeQueue = this.getWakeQueue();
        Iterator<Map.Entry<String, Runnable>> iterator = wakeQueue.entrySet().iterator();
        while (true) {
            while (iterator.hasNext()) {
                Map.Entry<String, Runnable> next = iterator.next();
                Runnable task = next.getValue();
                try {
                    task.run();
                } finally {
                    iterator.remove();
                }
            }
            Sleep.seconds(1);
        }
    }

    /**
     *  提交唤醒任务
     *
     *  @param  apiClientId 渠道客户端id
     *  @param  task        执行任务
     *  @author Pan
     */
    public static void submitWake(String apiClientId, Runnable task) {
        int getBind = new BigInteger(apiClientId).mod(new BigInteger(AVAILABLE_STR)).intValue();
        OrgWakeTask orgWakeTask = BIND_MAP.get(getBind);
        CacheFifo<String, Runnable> normalFlowPool = orgWakeTask.getWakeQueue();
        normalFlowPool.put(apiClientId, task);
    }
}